const path = require('path');
const { EventEmitter } = require('events');
const { Job } = require('./job');
const { Task } = require('./task');
const { DatasetListDB } = require('../../dbs/dbDatasetList');

class Pipeline extends EventEmitter {
    constructor(app) {
        super();
        this.app = app;
        this.db = app.pipe_db;
        this.broker = app.broker;
        this.config = null;
        this.nodes = null;
        this.nodeOutputs = null;

        // 用于存储等待结果的任务
        this.pendingTasks = {};
    }

    buildNodes(config, parentNode = null) {
        const rootNode = this.createNode(config, parentNode);
        this.nodes.set(rootNode.name, rootNode);

        if (config.nodes) {
            config.nodes.forEach((nodeConfig) => {
                const node = this.createNode(nodeConfig, rootNode);
                this.nodes.set(node.name, node);
                if (nodeConfig.nodes) {
                    this.buildNodes(nodeConfig, node);
                }
            });
        }

        this.initializeDefaultOutputs();
    }

    buildParser(configPath, config) {
        try {
            this.parser = null;

            // 获取解析器路径
            const parserPath = config.parser.path;
            if (!parserPath) {
                return;
            }

            // 构造完整路径
            const configDir = path.dirname(configPath);
            const parserFullPath = path.resolve(configDir, parserPath);
            const ParserClass = require(parserFullPath);
            if (ParserClass) {
                this.parser = new ParserClass(this.app);
            }
        } catch (error) {
            // 统一错误处理（可在此添加日志记录）
            throw new Error(`构建解析器失败: ${error.message}`);
        }
    }

    createNode(nodeConfig, parentNode) {
        const node = {
            name: nodeConfig.name,
            type: nodeConfig.type,
            worker: nodeConfig.worker,
            input: nodeConfig.input || [],
            output: nodeConfig.output || [],
            algo: nodeConfig.algo,
            nodes: nodeConfig.nodes || [],
            parent: parentNode
        };
        return node;
    }

    initializeDefaultOutputs() {
        for (const [nodeName, node] of this.nodes) {
            const defaultOutput = {};
            node.output.forEach((outputKey) => {
                defaultOutput[outputKey] = null;
            });
            this.nodeOutputs[nodeName] = defaultOutput;
        }
    }

    truncateInputData(inputData, maxLength = 50) {
        const truncateInputData = { ...inputData };
        const imagePath = truncateInputData.image_path;
        if (imagePath) {
            if (typeof imagePath === 'string' && imagePath.startsWith('data:image/')) {
                truncateInputData.image_path = imagePath.substring(0, maxLength) + '...';
            }
        }
        return truncateInputData;
    }

    async executeNode(node, inputData, job) {
        if (!node) {
            throw new Error(`Node is undefined. Check if the node exists in the pipeline.`);
        }

        let result;
        if (node.type === 'node') {
            result = await this.executeAlgoNode(node, inputData, job);
        } else if (node.type === 'serial') {
            result = await this.executeSerialNode(node, inputData, job);
        } else if (node.type === 'parallel') {
            result = await this.executeParallelNode(node, inputData, job);
        } else {
            throw new Error(`Unknown node type: ${node.type}`);
        }

        // console.log('\n## ======================================');
        // console.log('## Resolving inputs for node:', node.name);
        // console.log('## Node inputs:', node.input);
        // console.log(`## Input data:`, this.truncateInputData(inputData));
        // console.log(`## Node "${node.name}" outputs:`, result);
        // console.log('## ======================================\n');

        return result;
    }

    async executeAlgoNode(node, inputData, job) {
        // 创建task对象
        const taskId = await this.db.tasks.insert({
            job_id: job.id,
            name: node.name,
            input: JSON.stringify(inputData),
            algo: node.algo,
            status: 'pending',
            startTime: null,
            endTime: null
        });

        const task = new Task(taskId, job.job_id, node.name, inputData, node.algo);
        task.start();

        // 更新任务状态到数据库
        await this.db.tasks.update(task);

        // 更新当前正在执行的节点和任务
        job.setCurrentNode(node.name);
        job.setCurrentTask(task.taskId);
        await this.db.jobs.update(job);

        let result;
        try {
            // 解析输入数据
            const inputs = this.resolveInputs(node, inputData);
            // 执行异步远程任务
            const outputs = await this.callAlgo(node, inputs, job);
            // 解析输出数据
            result = this.resolveOutputs(node, outputs);

            // 更新节点输出
            this.updateNodeOutputs(node, result);

            // 标记任务为完成
            task.complete(JSON.stringify(result));
        } catch (error) {
            // 标记任务为失败
            task.fail(JSON.stringify({ error: error.message, stack: error.stack }));
            throw error;
        } finally {
            // 更新任务状态到数据库
            await this.db.tasks.update(task);
        }

        return result;
    }

    async executeSerialNode(node, inputData, job) {
        let currentData = { ...inputData };
        for (const childNode of node.nodes) {
            const childNodeInstance = this.nodes.get(childNode.name);
            if (!childNodeInstance) {
                throw new Error(`Child node ${childNode.name} not found.`);
            }
            const nodeOutput = await this.executeNode(childNodeInstance, currentData, job);

            const mergedOutput = Object.keys(nodeOutput).reduce((acc, key) => {
                const newKey = `${childNode.name}.${key}`;
                acc[newKey] = nodeOutput[key];
                return acc;
            }, {});

            currentData = { ...currentData, ...mergedOutput };
        }
        return this.resolveOutputs(node, currentData);
    }

    async executeParallelNode(node, inputData, job) {
        const promises = node.nodes.map((childNode) => {
            const childNodeInstance = this.nodes.get(childNode.name);
            if (!childNodeInstance) {
                throw new Error(`Child node ${childNode.name} not found.`);
            }
            return this.executeNode(childNodeInstance, inputData, job);
        });
        const results = await Promise.all(promises);

        const outputData = results.reduce((acc, result, index) => {
            const childNode = node.nodes[index];
            Object.keys(result).forEach((key) => {
                const newKey = `${childNode.name}.${key}`;
                acc[newKey] = result[key];
            });
            return acc;
        }, {});
        return this.resolveOutputs(node, outputData);
    }

    resolveInputs(node, inputData) {
        return node.input.reduce((acc, input) => {
            // 处理对象形式的输入配置
            if (typeof input === 'object' && input.name && input.from) {
                const [nodeName, outputKey] = input.from.split('.');
                if (!this.nodeOutputs[nodeName] || !this.nodeOutputs[nodeName][outputKey]) {
                    console.warn(`Using default value for input: ${input.from}`);
                    acc[input.name] = null;
                } else {
                    acc[input.name] = this.nodeOutputs[nodeName][outputKey];
                }
            }
            // 处理字符串形式的输入配置
            else if (typeof input === 'string') {
                if (input.includes('.')) {
                    const [nodeName, outputKey] = input.split('.');
                    if (!this.nodeOutputs[nodeName] || !this.nodeOutputs[nodeName][outputKey]) {
                        console.warn(`Using default value for input: ${input}`);
                        acc[input] = null;
                    } else {
                        acc[input] = this.nodeOutputs[nodeName][outputKey];
                    }
                } else {
                    if (!(input in inputData)) {
                        throw new Error(`Input dependency not found: ${input}`);
                    }
                    acc[input] = inputData[input];
                }
            }
            return acc;
        }, {});
    }

    resolveOutputs(node, outputData) {
        const output = {};
        node.output.forEach((outputKey) => {
            const [nodeName, key] = outputKey.includes('.') ? outputKey.split('.') : [node.name, outputKey];
            if (node.type === 'node') {
                output[outputKey] = outputData[outputKey] || null;
            } else {
                output[outputKey] = this.nodeOutputs[nodeName][key] || null;
            }
        });
        return output;
    }

    updateNodeOutputs(node, outputData) {
        node.output.forEach((outputKey) => {
            if (!this.nodeOutputs[node.name]) {
                this.nodeOutputs[node.name] = {};
            }
            this.nodeOutputs[node.name][outputKey] = outputData[outputKey];
        });
    }

    async callAlgo(node, inputs, job) {
        // 发送任务到 Broker
        const taskId = `task_${Date.now()}_${Math.random().toString(36).substring(7)}`;
        await this.broker.publish(node.algo, node.worker, {
            jobId: job.id,
            taskId,
            inputs,
            nodeName: node.name
        });

        // 返回一个 Promise，等待任务结果
        return new Promise((resolve, reject) => {
            this.pendingTasks[taskId] = { resolve, reject };

            // 设置超时机制，避免任务卡死
            setTimeout(() => {
                if (this.pendingTasks[taskId]) {
                    delete this.pendingTasks[taskId];
                    reject(new Error(`Task ${taskId} timed out after 10 seconds`));
                }
            }, 10000); // 10秒超时
        });
    }

    // 读取配置文件，创建pipeline的node节点
    loadConfig(configPath, config) {
        this.config = config;
        this.nodes = new Map();
        this.nodeOutputs = {};
        this.buildParser(configPath, config);
        this.buildNodes(this.config);
    }

    ////////////////////////////////////////////////////////
    // 实际执行pipeline的接口
    async execute(inputData, job) {
        // 正式执行pipeline
        try {
            const rootNode = this.nodes.get(this.config.name);
            if (!rootNode) {
                throw new Error(`Root node ${this.config.name} not found.`);
            }
            const result = await this.executeNode(rootNode, inputData, job);
            // console.log('======================================');
            // console.log('Final node outputs:', this.nodeOutputs);
            // console.log('======================================\n');
            return result;
        } catch (error) {
            console.log(error);
            job.fail(JSON.stringify(error));
            await this.db.jobs.update(job);
            this.emit('failed', {
                jobId: job.id,
                input: inputData,
                error: error.message
            });
            return null;
        }
    }

    // 注册任务结果处理器
    task_callback = async (message) => {
        console.log(`Task result called, message = `, message);
        const { taskId, result } = message;
        if (this.pendingTasks[taskId]) {
            this.pendingTasks[taskId].resolve(result);
            delete this.pendingTasks[taskId];
        }
    }

    // 启动单独一条数据
    async run(inputData) {
        // 创建任务并订阅结果
        const jobId = await this.db.jobs.insert({
            input: JSON.stringify(inputData),
            config: this.config.name
        });
        const job = new Job(jobId, this.config.name, inputData);
        job.start();
        await this.db.jobs.update(job);
        this.broker.subscribe(`task_result`, this.task_callback);

        // 执行任务
        this.emit('percent', {
            jobId: jobId,
            percent: 0
        });
        let result = await this.execute(inputData, job);
        let labelResult = null;
        if (this.parser) {
            labelResult = this.parser.convertResult(result);
        }
        this.emit('done', {
            jobId: job.id,
            input: inputData,
            result: result,
            labelResult: labelResult
        });
        this.emit('percent', {
            jobId: jobId,
            percent: 100
        });

        // 取消订阅并结束任务
        job.complete(JSON.stringify([]));
        await this.db.jobs.update(job);
        this.broker.unsubscribe('task_result', this.task_callback);
    }

    // 对于batch模式，根据inputInfo来更新数据
    genInputData(inputData, inputInfo, rawData) {
        const input = { ...inputData };
        for (const [key, value] of Object.entries(inputInfo)) {
            if (value.rawData) {
                input[key] = rawData;
            } else {
                input[key] = value.value;
            }
        }
        return input;
    }

    async runBatch(inputData, queueName) {
        console.log(`runBatch inputData = `, inputData);

        // 定义正则表达式，匹配 dataset://xxx/xxx 格式
        const datasetRegex = /^dataset:\/\/([^\/]+)\/(.+)$/;

        // 遍历 inputData 的每个字段
        const inputInfo = {};
        let machine, datasetId;
        for (const [key, value] of Object.entries(inputData)) {
            if (typeof value === 'string') {
                const match = value.match(datasetRegex);
                if (match) {
                    // 如果匹配成功，解析出对应的部分
                    const [, _machine, _datasetId] = match;
                    machine = _machine;
                    datasetId = _datasetId;
                    inputInfo[key] = {
                        rawData: true,
                        value
                    };
                    continue
                }
            }
            inputInfo[key] = {
                rawData: false,
                value
            };
        }
        console.log(`machine = ${machine}, datasetId = ${datasetId}, inputInfo = `, inputInfo);

        // 从数据库中读取数据集
        let isValidID = await this.app.db.datasets.hasDataset(datasetId);
        if (!isValidID) {
            return { code: -2, msg: `Dataset ID ${datasetId} does not exist!` };
        }
        const dataset_info = await this.app.db.datasets.getDatasetInfo(datasetId);
        const dataset_dir = dataset_info.dataset_dir;

        // 打开详情数据库，获取对应的列表信息
        let dblist = this.app.list_dbs[datasetId];
        if (!dblist) {
            this.app.list_dbs[datasetId] = new DatasetListDB(this.app, datasetId);
            dblist = this.app.list_dbs[datasetId];
        }

        // 创建任务并订阅结果
        // 创建任务并订阅结果
        const jobId = await this.db.jobs.insert({
            input: JSON.stringify(inputData),
            config: this.config.name
        });
        const job = new Job(jobId, this.config.name, inputData);
        job.start();
        await this.db.jobs.update(job);
        this.broker.subscribe(`task_result`, this.task_callback);

        // 遍历所有页的数据
        let page = 0, page_size = 100;
        let pageData = await dblist.pairs.getPairsPageWithTotal(dataset_dir, page, page_size);
        let processedRecords = 0;
        const totalPages = pageData.totalPages;
        const totalRecords = pageData.totalRecords;

        // 更新数据集的状态
        await this.app.db.datasets.updateDataset({ id: datasetId, status: queueName, percent: 0 });

        // 处理数据集中的每一个数据记录
        for (let currentPage = 0; currentPage < totalPages; currentPage++) {
            console.log(`正在处理第 ${currentPage + 1} 页，共 ${totalPages} 页`);
            // 获取当前页的数据
            if (currentPage > 0) {
                pageData = await dblist.pairs.getPairsPageWithTotal(dataset_dir, currentPage, page_size);
            }

            // 遍历当前页的每一条记录
            for (const record of pageData.data) {
                const image_path = `file://${record.image_path}`;
                const input = this.genInputData(inputData, inputInfo, image_path);

                // 执行任务并保存标注的结果
                let result = await this.execute(input, job);
                if (this.parser) {
                    await this.parser.saveResult(dataset_info, record, result);
                }

                // 进度+1
                processedRecords += 1;
                let percent = (processedRecords * 100 / totalRecords).toFixed(2);
                this.emit('percent', {
                    jobId: jobId,
                    datasetId: datasetId,
                    percent: percent
                });
                await this.app.db.datasets.updateDataset({ id: datasetId, status: queueName, percent: percent });
            }
        }

        // 取消订阅并更新任务状态
        job.complete(JSON.stringify([]));
        await this.db.jobs.update(job);
        await this.app.db.datasets.updateDataset({ id: datasetId, status: 'idle', percent: 0 });
        this.emit('done', {
            jobId: job.id,
            input: inputData,
            result: 'done'
        });

        this.broker.unsubscribe('task_result', this.task_callback);
    }
}

module.exports = { Pipeline };
